package com.ttdevs.socket.tcp;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

import com.ttdevs.socket.ISocket;
import com.ttdevs.socket.util.BaseThread;
import com.ttdevs.socket.util.IPConfig;
import com.ttdevs.socket.util.SimpleSocket;
import com.ttdevs.socket.util.SocketConfig;

public class SocketTCPImp extends BaseThread implements ISocket {

	private TCPSender tcpSender;
	private TCPReceiver tcpReceiver;
	private BlockingQueue<byte[]> senderQueue;
	private BlockingQueue<byte[]> receiverQueue;
	private OnSocketListener mListener;
	private SimpleSocket mSocket;

	public SocketTCPImp() {
		senderQueue = new ArrayBlockingQueue<byte[]>(SocketConfig.SENDQUEUESIZE);
		receiverQueue = new ArrayBlockingQueue<byte[]>(SocketConfig.RECEIVEQUEUESIZE);
	}

	@Override
	public boolean connect(IPConfig ipConfig, OnSocketListener listener) {
		if (listener == null) {
			return false;
		}
		mListener = listener;

		mSocket = new SimpleSocket();

		if (!mSocket.create(ipConfig)) {
			return false;
		}

		tcpSender = new TCPSender(senderQueue, this, mSocket.getOutputStream());
		tcpSender.start();

		tcpReceiver = new TCPReceiver(receiverQueue, this, mSocket.getInputStream());
		tcpReceiver.start();
		
		this.start();
		return true;
	}

	@Override
	public void close(boolean isNotify, boolean isByServer) {
		mSocket.close();
		if (isNotify) {
			mListener.onClose(isByServer, this);
		}
	}

	@Override
	public boolean sendData(byte[] data) {
		if (data != null) {
			try {
				senderQueue.put(data);
			} catch (InterruptedException e) {
				return false;
			}
			return true;
		}
		return false;
	}

	@Override
	public void execute() {
		try {
			byte[] data = receiverQueue.take();
			if (data != null) {
				mListener.onRead(data, this);
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
			close(true, false);
			quit();
		}
	}
}
